package org.ala.distributed_transaction_commons.communication.coordinator.actions;

import org.ala.distributed_transaction_commons.DistributedTransactionStatus;
import org.ala.distributed_transaction_commons.communication.Message;
import org.ala.distributed_transaction_commons.communication.coordinator.AbstractCoordonatorAction;
import org.ala.distributed_transaction_commons.logs.LoggerFactory;
import org.ala.distributed_transaction_commons.redis.datas.TransactionContext;
import org.slf4j.Logger;

import redis.clients.jedis.Pipeline;


/**
 * 协调者收到can_commit消息做出的响应
 * <p>	检测事务集群中所有参与者是否都是can_commit状态。若是，则向参与者发送pre_commit消息，修改redis中事务状态为pre_commit
 * 
 * @author 骆毅(ala)
 * @date 2024年2月18日
 */
public class CanCommitAction extends AbstractCoordonatorAction {

	
	static Logger log = LoggerFactory.getLogger("distributed_transaction/cooridinator", "cooridinator");
	
	
	@Override
	public String type() {
		return Message.MSG_TYPE_TO_COORDINATOR_CAN_COMMIT;
	}

	@Override
	protected void doExecute(TransactionContext context) {
		//	检测事务集群中所有参与者是否都>=can_commit状态（兼容有人直接跳到后面步骤的情况）
		boolean isAllCancommit = context.getParticipantSet().stream().map(t -> DistributedTransactionStatus.CAN_COMMIT.lte(t.getStatus())).reduce(true, (t1, t2) -> t1 && t2);
		
		//	若是，则向参与者发送pre_commit消息，修改redis中事务状态为pre_commit
		int participantCount = context.getParticipantSet().size();
		if (isAllCancommit && participantCount >= context.getMinParticipantCountCanCommit()) {
			distributedTransactionRedisTools.withJedis(jedis -> {
				Pipeline tx = jedis.pipelined();
				participantPublisher.sendParticipantPreCommit(tx, context.getBiz(), context.getId());
				distributedTransactionRedisTools.updateContextStatus(tx, context.getBiz(), context.getId(), DistributedTransactionStatus.PRE_COMMIT);
				tx.sync();
				return null;
			});
			log.info("[CanCommitAction doExecute] 协调者收到can_commit消息，所有参与者均处于can_commit状态，广播pre_commit消息 biz:" + context.getBiz() + " txId:" + context.getId());
		} else {
			log.info("[CanCommitAction doExecute] 协调者收到can_commit消息，但当前还有参与者未处于can_commit状态，或参与者人数不足，继续等待 biz:" + context.getBiz() + " txId:" + context.getId() + " participantCount:" + participantCount + " minParticipantCount:" + context.getMinParticipantCountCanCommit());
		}
	}
}
